package com.zhl.weibo.util;

import com.zhl.weibo.bean.KeyValBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @program: demos
 * @description:
 * @author: 刘振华
 * @create: 2020-11-05 19:14
 **/
public class HBaseUtil {
	private static Configuration conf = HBaseConfiguration.create();

	public static void deleteData(String tableName, String rowKey, String cf, String column){
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			table = conn.getTable(tableObj);
			Delete target = new Delete(Bytes.toBytes(rowKey));
			if(cf!=null&&!"".equals(cf)){
				if(column!=null&&!"".equals(column)){
					target.addColumns(Bytes.toBytes(cf),Bytes.toBytes(column));
				}else {
					target.addFamily(Bytes.toBytes(cf));
				}
			}
			table.delete(target);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static List<KeyValBean> scan(String tableName, String cf, String column,Integer version, String startRow, String stopRow){
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			table = conn.getTable(tableObj);
			Scan scan = new Scan();//scan可以传范围
			if(cf!=null && !"".equals(cf)){
				if(column!=null&&!"".equals(column)){
					scan.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));
				}else{
					scan.addFamily(Bytes.toBytes(cf));
				}
			}
			if(startRow!=null&&!"".equals(startRow)){
				scan.withStartRow(Bytes.toBytes(startRow));
			}

			if(stopRow!=null&&!"".equals(stopRow)){
				scan.withStopRow(Bytes.toBytes(stopRow));
			}
			if(version != null && version >0) {
				scan.readVersions(version);
			}
			final List<KeyValBean> rbList = new ArrayList<>();
			for(Result rs : table.getScanner(scan)){
				rs.listCells().forEach(c-> rbList.add(KeyValBean.builder()
					.rowKey(Bytes.toString(CellUtil.cloneRow(c)))
					.cf(Bytes.toString(CellUtil.cloneFamily(c)))
					.colName(Bytes.toString(CellUtil.cloneQualifier(c)))
					.value(Bytes.toString(CellUtil.cloneValue(c)))
					.timestamp(c.getTimestamp())
					.build()));
			}
			return rbList;
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return null;
	}

	//获取数据
	public static List<KeyValBean> get(String tableName, String rowKey, String cf, String column,Integer version){
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			table = conn.getTable(tableObj);
			if(cf!=null && !"".equals(cf)){
				if(column!=null && !"".equals(column)){
					get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(column));
				}else{
					get.addFamily(Bytes.toBytes(cf));
				}
			}
			if(null!= version && version>0) {
				get.readVersions(version);
			}
			final List<KeyValBean> rbList = new ArrayList<>();
			table.get(get).listCells().forEach(
				c-> rbList.add(KeyValBean
					.builder()
					.rowKey(Bytes.toString(CellUtil.cloneRow(c)))
					.cf(Bytes.toString(CellUtil.cloneFamily(c)))
					.colName(Bytes.toString(CellUtil.cloneQualifier(c)))
					.value(Bytes.toString(CellUtil.cloneValue(c)))
					.timestamp(c.getTimestamp())
					.build()));
			return rbList;
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return null;
	}

	public static void put(String tableName, Collection<KeyValBean> collection){
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			table = conn.getTable(tableObj);
			for(KeyValBean bean : collection){
				Put p = new Put(Bytes.toBytes(bean.getRowKey()));
				p.addColumn(Bytes.toBytes(bean.getCf()), Bytes.toBytes(bean.getColName()),
					Bytes.toBytes(bean.getValue()));
				table.put(p);
			}
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static void put(String tableName, String rowKey, String cf, String column, String value){
		Connection conn = null;
		Table table = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			TableName tableObj = TableName.valueOf(tableName);
			table = conn.getTable(tableObj);
			Put p = new Put(Bytes.toBytes(rowKey));
			p.addColumn(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(value));
			table.put(p);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != table) {
					table.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static void dropTable(String tableName){
		Connection conn = null;
		Admin admin = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
			boolean exist = admin.tableExists(TableName.valueOf(tableName));
			if(!exist){
				System.out.println(String.format("表%s不存在",tableName));
				return;
			}
			TableName table = TableName.valueOf(tableName);
			admin.disableTable(table);
			admin.deleteTable(table);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != admin) {
					admin.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	public static boolean isTableExist(String tableName){
		Connection conn = null;
		Admin admin = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
			return admin.tableExists(TableName.valueOf(tableName));
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != admin) {
					admin.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return false;
	}
	public static void creteTable(String tableName,Integer version,String... cfs){
		Connection conn = null;
		Admin admin = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
			boolean exist = admin.tableExists(TableName.valueOf(tableName));
			if(exist){
				System.out.println(String.format("表%s已存在",tableName));
				return;
			}
			if(cfs.length < 1){
				System.out.println("列族不能为空!");
				return;
			}
			if(version == null || version == 0){
				version = 3;
			}
			final int v = version;
			List<ColumnFamilyDescriptor> cfDescriptorColl = Arrays.asList(cfs)
				.stream().map(cf->{
					ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder
						.newBuilder(Bytes.toBytes(cf))
						.setMaxVersions(v)
						.build();
					return cfd;
				})
				.collect(Collectors.toList());
			TableDescriptor descriptor = TableDescriptorBuilder
				.newBuilder(TableName.valueOf(tableName))
				.setColumnFamilies(cfDescriptorColl)
				.build();
			admin.createTable(descriptor);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != admin) {
					admin.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	public static void createNamespace(String namespace){
//		conf.set("hbase.zookeeper.quorum","C111908191,C111908192,C111908193");
		Connection conn = null;
		Admin admin = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
			admin.createNamespace(NamespaceDescriptor.create(namespace).build());
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != admin) {
					admin.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	public static void dropNamespace(String namespace){
//		conf.set("hbase.zookeeper.quorum","C111908191,C111908192,C111908193");
		Connection conn = null;
		Admin admin = null;
		try {
			conn = ConnectionFactory.createConnection(conf);
			admin = conn.getAdmin();
			admin.deleteNamespace(namespace);
		} catch (IOException e) {
			e.printStackTrace();
		}finally {
			try {
				if(null != admin) {
					admin.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				if(null != conn) {
					conn.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}
